﻿using System.Text;
using HuaskyWebAPI.Common.Util;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace HuaskyWebAPI.Common.Rabbitmq;

public static class MQUtil
{
    public static void QueueDeclare(this IModel channel, string queue)
    {
        channel.QueueDeclare(queue, true, false, false, null);
    }

    public static void Publish(this IModel channel, string exchange, string route, object item)
    {
        var content = JsonConvert.SerializeObject(item);
        var bytes = Encoding.UTF8.GetBytes(content);
        channel.BasicPublish(exchange, route, null, bytes);
    }


    public static void Subscribe(this IModel model, string queue, ushort prefetchCount, Func<MQPara, Task> action)
    {
        prefetchCount = Math.Max(prefetchCount, (ushort)1);
        model.BasicQos(0, prefetchCount, false);

        var consumer = new EventingBasicConsumer(model);

        //接收到消息时执行的任务
        consumer.Received += (m, ea) =>
        {
            onMqMessage(model, ea, action);
        };

        model.BasicConsume(queue: queue, autoAck: false, consumer: consumer);
    }


    /// <summary>
    /// 批量处理消息，消息处理完或到达最大次数后返回，不能和Subscribe同时使用
    /// </summary>
    public static async Task<bool> BasicConsume<T>(this IModel model, string queue, Func<T, Task> action)
    {
        var msg = model.BasicGet(queue, false);
        if (msg == null)
        {
            return false;
        }

        try
        {
            var para = new MQPara(msg);
            var content = para.ToObject<T>();
            await action(content);
        }
        catch (Exception e)
        {
            LoggerHelper.Error(e);
        }
        finally
        {
            model.BasicAck(msg.DeliveryTag, false);
        }

        return true;
    }

    /// <summary>
    /// 获取消息体
    /// </summary>
    internal static T GetContent<T>(this BasicDeliverEventArgs args)
    {
        return GetContent<T>(args.Body);
    }


    /// <summary>
    /// 获取消息体
    /// </summary>
    internal static T GetContent<T>(ReadOnlyMemory<byte> body)
    {
        if (typeof(T) == typeof(string))
        {
            return (T)((object)Encoding.UTF8.GetString(body.Span));
        }
        else
        {
            var content = Encoding.UTF8.GetString(body.Span);
            return JsonConvert.DeserializeObject<T>(content);
        }
    }


    static void onMqMessage(IModel model, BasicDeliverEventArgs eventArgs, Func<MQPara, Task> action)
    {
        var ctx = new MQPara(eventArgs);
        try
        {
            var completion = action(ctx);
            completion.ContinueWith(_ => ack());
        }
        catch (Exception e)
        {
            //异常消息在业务中处理，正常情况下不应该进入此流程
            LoggerHelper.Error(e, "receive message fail");
            ack();
        }

        //消息确认
        void ack()
        {
            try
            {
                model.BasicAck(eventArgs.DeliveryTag, false);
            }
            catch (Exception e)
            {
                LoggerHelper.Error(e, "Ack mq msg fail");
            }
        }
    }
}


public class MQPara
{
    public IDictionary<string, object> Headers { get; }

    public ReadOnlyMemory<byte> Body { get; }

    internal MQPara(BasicDeliverEventArgs args)
        : this(args.Body, args.BasicProperties)
    {
    }

    internal MQPara(BasicGetResult result)
        : this(result.Body, result.BasicProperties)
    {
    }

    internal MQPara(ReadOnlyMemory<byte> body, IBasicProperties properties)
    {
        this.Body = body;
        this.Headers = properties.Headers;
    }




    public string Content => Encoding.UTF8.GetString(Body.Span);

    public T ToObject<T>()
    {
        return MQUtil.GetContent<T>(Body);
    }
}